package com.yangye.otherdemo.mythreadpoolexecutor;

import javafx.concurrent.Worker;

import java.util.HashSet;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Handler;

/**
 * @author yangye
 * @date 2020/08/18
 */
public class YangyeThreadPoolExecutor implements YangyeExecutorService {

	// 线程池大小
	private int poolSize;

	// 活动线程数
	private AtomicInteger ctl = new AtomicInteger(0);

	// 拒绝策略
	private PolicyHandler policyHandler;

	// 等待队列
	private BlockingQueue<Runnable> workQueue;

	// 线程池状态
	private boolean isShutdown = false;

	// 任务完成数量
	private long completedTaskCount;

	private ReentrantLock mainLock = new ReentrantLock();

	private final HashSet<Worker> workers = new HashSet<Worker>();

	public YangyeThreadPoolExecutor(int poolSize, PolicyHandler policyHandler, BlockingQueue<Runnable> workQueue) {
		this.poolSize = poolSize;
		this.policyHandler = policyHandler;
		this.workQueue = workQueue;
	}

	@Override
	public void execute(Runnable task) {
		if (task == null) {
			throw new RuntimeException("任务不能为空");
		}
		if (isShutdown) {
			throw new RuntimeException("线程池已经停止，不能继续添加任务");
		}

		int wc = ctl.get();
		if (wc < poolSize) {
			// 可以继续添加任务
			if (addWorker(task)) {
				return;
			}
		} else if (workQueue.offer(task)) {

		} else {
			policyHandler.reject(task, this);
		}


	}

	private boolean addWorker(Runnable task) {

		boolean workerAdded = false;
		boolean workerStarted = false;

		ctl.getAndIncrement();

		Worker worker = new Worker(task);
		Thread t = worker.thread;

		if (t != null) {
			mainLock.lock();
			try {
				if (!isShutdown) {// 检查线程池是否停止
					if (t.isAlive()) {// 线程已经start，不能再次start
						throw new IllegalThreadStateException();
					}
					workers.add(worker);
					workerAdded = true;
				}
			} finally {
				mainLock.unlock();
			}
			if (workerAdded) {
				t.start();
				workerStarted = true;
			}
		}
		return workerStarted;
	}

	@Override
	public void shutdown() {
		mainLock.lock();
		try {
			isShutdown = true;
			for (Worker w : workers) {
				Thread t = w.thread;
				if (!t.isInterrupted() && w.tryLock()) {
					try {
						t.interrupt();
					} catch (Exception e) {

					} finally {
						w.unlock();
					}
				}
			}
		} finally {
			mainLock.unlock();
		}
	}

	@Override
	public int getPoolSize() {
		return poolSize;
	}

	@Override
	public Runnable getTask() {
		return workQueue.poll();
	}

	static AtomicInteger atomic = new AtomicInteger();
	class Worker extends ReentrantLock implements Runnable{

		volatile long completedTask;
		final Thread thread;
		Runnable firstTask;

		public Worker(Runnable r){
			this.firstTask = r;
			this.thread = new Thread(this,"thread-name-"+atomic.incrementAndGet());
		}

		public void run() {
			runWorker(this);
		}
	}

	private void runWorker(Worker worker) {
		Thread wt = Thread.currentThread();
		Runnable task = worker.firstTask;
		worker.firstTask = null;
		boolean completedAbruptly = true;
		try {
			while (task != null || (task=getTask())!=null){
				worker.lock();
				if (isShutdown && !wt.isInterrupted()){
					wt.interrupt();
				}
				try {
					task.run();
				} finally {
					task = null;
					worker.completedTask++; //当前线程完成的任务数
					worker.unlock();
				}
			}
			completedAbruptly = false;
		} finally {
			processWorkerExit(worker,completedAbruptly);
		}

	}

	private void processWorkerExit(Worker worker, boolean completedAbruptly) {
		if(completedAbruptly)
			ctl.decrementAndGet();

		final ReentrantLock mainLock = this.mainLock;
		mainLock.lock();
		try {
			completedTaskCount += worker.completedTask;
			workers.remove(worker);
		} finally {
			mainLock.unlock();
		}
		if(completedAbruptly && !workQueue.isEmpty()){
			addWorker(null);
		}
	}
}
